package com.yingsheng.bi.hks.repository.redis;

import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.geo.*;
import org.springframework.data.redis.connection.RedisGeoCommands;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;

import java.util.*;

import org.springframework.data.redis.connection.RedisGeoCommands.*;

/**
 * redis地理空间索引服务
 *
 * @author root
 */
@Slf4j
@Service
public class GeoRedisService {

    private final RedisTemplate<Object, Object> redisTemplate;

    @Autowired
    public GeoRedisService(RedisTemplate<Object, Object> redisTemplate) {
        this.redisTemplate = redisTemplate;
    }

    /**
     * 添加地理位置信息(单个)
     */
    public void add(String key, String member, double longitude, double latitude) {
        redisTemplate.opsForGeo().add(key, new Point(longitude, latitude), member);
    }

    /**
     * 添加地理位置信息(批量)
     */
    public Long add(String key, Map<Object, Point> locations) {
        if (StringUtils.hasText(key) && !CollectionUtils.isEmpty(locations)) {
            return redisTemplate.opsForGeo().add(key, locations);
        }
        return 0L;
    }

    /**
     * 获取指定成员的位置信息。
     */
    public Point position(String key, String member) {
        if (StringUtils.hasText(key) && StringUtils.hasText(member)) {
            List<Point> points = redisTemplate.opsForGeo().position(key, member);
            return CollectionUtils.isEmpty(points) ? null : points.get(0);
        }
        return null;
    }

    /**
     * 批量获取多个成员的位置信息。
     */
    public Map<String, Point> position(String key, Collection<String> members) {
        Map<String, Point> result = new HashMap<>();
        if (StringUtils.hasText(key) && !CollectionUtils.isEmpty(members)) {
            String[] numberArray = members.toArray(new String[0]);
            List<Point> pointList = redisTemplate.opsForGeo().position(key, numberArray);
            if (!CollectionUtils.isEmpty(pointList)) {
                for (int i = 0; i < members.size(); i++) {
                    result.put(numberArray[i], pointList.get(i));
                }
            }
        }
        return result;
    }


    /**
     * 获取两个位置之间的距离
     */
    public Double distance(String key, String member1, String member2, RedisGeoCommands.DistanceUnit unit) {
        return Objects.requireNonNull(redisTemplate.opsForGeo().distance(key, member1, member2, unit)).getValue();
    }

    /**
     * 删除指定键中的一个或多个地理空间成员。
     */
    public Long remove(String key, Collection<String> members) {
        if (StringUtils.hasText(key) && !CollectionUtils.isEmpty(members)) {
            return redisTemplate.execute((RedisCallback<Long>) connection -> {
                byte[] rawKey = redisTemplate.getStringSerializer().serialize(key);
                byte[][] rawMembers = members.stream()
                        .map(member -> redisTemplate.getStringSerializer().serialize(member))
                        .toArray(byte[][]::new);
                return connection.geoRemove(rawKey, rawMembers);
            });
        }
        return 0L;

    }

    /**
     * 查询距离中心点范围内的成员列表
     *
     * @param key    redis的key
     * @param center 中心点坐标
     * @param radius 距离中心半径
     * @param count  返回最大数量
     * @param sort   排序;0-升序,1-降序,其他-不排序;不送的话默认不是升序也不是降序,而是按redis内部的处理顺序返回
     * @return 返回距离中心半径范围内成员列表,其中GeoResult=距离+GeoLocation, GeoLocation=坐标+成员名称
     */
    public List<GeoResult<GeoLocation<Object>>> radius(String key, Point center, Distance radius, int count, int sort) {
        if (StringUtils.hasText(key) && center != null && radius != null) {
            Circle within = new Circle(center, radius);
            RedisGeoCommands.GeoRadiusCommandArgs args = RedisGeoCommands.GeoRadiusCommandArgs.newGeoRadiusArgs();
            switch (sort) {
                case 0:
                    args.sortAscending();
                    break;
                case 1:
                    args.sortDescending();
                    break;
            }
            if (count > 0) {
                args.limit(count);
            }
            GeoResults<GeoLocation<Object>> results = redisTemplate.opsForGeo().radius(key, within, args);
            if (results != null) {
                return results.getContent();
            }
        }
        return new ArrayList<>();
    }

    /**
     * 查询距离中心点几公里内的成员列表
     *
     * @param key    redis的key
     * @param center 中心点坐标
     * @param radius 距离中心半径多少公里
     * @return 返回距离中心半径范围内成员列表,其中GeoResult=距离+GeoLocation, GeoLocation=坐标+成员名称
     */
    public List<GeoResult<GeoLocation<Object>>> radius(String key, Point center, double radius) {
        return radius(key, center, new Distance(radius, Metrics.KILOMETERS), -1, 0);
    }
}